package com.qhee.spark

import org.apache.http.client.config.RequestConfig
import org.apache.http.client.methods.{HttpGet, HttpRequestWrapper, HttpUriRequest}
import org.apache.http.impl.client.HttpClients
import org.apache.http.protocol.{BasicHttpContext, HttpCoreContext}
import org.apache.http.util.EntityUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.htmlcleaner.{HtmlCleaner, TagNode}

import scala.collection.mutable.ArrayBuffer


object SearchPartner {
  // val hdfsPartner = "hdfs://10.19.2.28:9000/spark/data/table/customer/" ;
  val hdfsPartner = "hdfs://10.19.2.28:9000/spark/data/table/customer";
  val sparkMaster = "spark://10.19.2.28:7077";
  val jarsList = List("D:\\temp\\spark\\alan_practice_jar\\alan-practice.jar");
  val seList: List[Map[String, String]] = List(
    Map("urlTpl" -> "http://www.baidu.com/s?wd=$keyword&pn=$pageno", "linkout" -> "//div[@class='result c-container']/h3/a/@href", "text" -> "//body/*"),
    Map("urlTpl" -> "http://news.baidu.com/ns?word=$keyword&pn=$pageno", "linkout" -> "", "text" -> "//body/*"),
    Map("urlTpl" -> "http://tieba.baidu.com/f?kw=$keyword&pn=$pageno", "linkout" -> "", "text" -> "//body/*"),
    Map("urlTpl" -> "http://zhidao.baidu.com/search?word=$keyword&pn=$pageno", "linkout" -> "", "text" -> "//body/*")
  );

  def main(args: Array[String]) = {
    doSearch("前海股权交易中心");
    //runSparkJob();
  }

  def runSparkJob() = {
    val conf = new SparkConf().setAppName("SearchPartner--Alan").setMaster(sparkMaster).setJars(jarsList);
    val spark = new SparkContext(conf);
    val partner = spark.textFile(hdfsPartner);
    partner.map(line => {
      line.split(",").head.replaceAll("'", "")
    }).top(10).foreach(kw => {
      println("########kw=" + kw);
      doSearch(kw);
    });
    //spark.parallelize(1 to 3 ,3).fl
    spark.stop();
  }

  def doSearch(kw: String) = {
    val depth = 2;
    for (se <- seList) {
      System.out.println("#######se=" + se);
      val linkin = se("urlTpl").replace("$keyword", kw).replace("$pageno", "0");
      crawl(linkin, depth, se);
    }
  }

  def crawl(linkin: String, depth: Int, setting: Map[String, String]): Unit = {
    println("crawl start ,depth=" + depth + ",linkin=" + linkin);
    var htmlInfo: Map[String, Object] = null;
    try {
      htmlInfo = grabPageHTML(linkin, setting);
    } catch {
      case unknown: Throwable => {
        println("crawl end,depth=" + depth + ",linkin=" + linkin + ",Unknown exception " + unknown.getMessage());
        return;
      }
    }

    if (depth > 1) {
      val nodes = htmlInfo("linkout").asInstanceOf[ArrayBuffer[String]];
      for (node <- nodes) {
        crawl(node, depth - 1, setting);
      }
    } else {
    }
    println("crawl end,depth=" + depth + ",linkin=" + linkin);

    println(htmlInfo);
  }

  // 获取一个HTML页面的内容，一个简单的get应用
  def grabPageHTML(linkin: String, setting: Map[String, String]): Map[String, Object] = {
    var defaultRequestConfig = RequestConfig.custom()
      .setSocketTimeout(10000).build();
    val httpget = new HttpGet(linkin);
    httpget.setConfig(defaultRequestConfig);
    val httpContext = new BasicHttpContext();
    val httpclient = HttpClients.createDefault();
    val response = httpclient.execute(httpget, httpContext);
    val request = httpContext.getAttribute(HttpCoreContext.HTTP_REQUEST).asInstanceOf[HttpUriRequest];
    //pageUrl is always correct , even if there is a redirect
    val pageUrl = request.asInstanceOf[HttpRequestWrapper].getOriginal().asInstanceOf[HttpGet].getURI().toString();
    val entity = response.getEntity();
    val html = EntityUtils.toString(entity, "GBK");
    EntityUtils.consume(entity);
    httpget.releaseConnection();


    val linkout = new ArrayBuffer[String]();
    val cleaner = new HtmlCleaner();
    val doc = cleaner.clean(html);
    var nodes = doc.evaluateXPath(setting("linkout"));
    for (node <- nodes) {
      val nextUrl = node.toString();
      linkout.append(nextUrl);
    }

    val text = new ArrayBuffer[String]();
    nodes = doc.evaluateXPath(setting("text"));
    for (node <- nodes) {
      val tagNode = node.asInstanceOf[TagNode];
       text.append(tagNode.getText().toString());
    }


    println("#######page end, linkin=" + linkin + ",pageUrl=" + pageUrl);
    //System.out.println(html);
    //return html ;

    Map("pageUrl" -> pageUrl, "linkin" -> linkin, "html" -> html, "linkout" -> linkout,"text"->text);
    // Map();
  }


}
